package com.shujia.core

import org.apache.spark.{SparkConf, SparkContext}

import java.sql.{Connection, DriverManager, PreparedStatement}

object Demo27ForeachPartition {
  def main(args: Array[String]): Unit = {
    // 将最终结果保存到外部系统
    val conf: SparkConf = new SparkConf()
    conf.setAppName("Demo27ForeachPartition") // 设置任务名
    conf.setMaster("local") // 设置Spark的运行方式
    // 创建SparkContext --> Spark程序的入口
    val sc: SparkContext = new SparkContext(conf)

    // 链式调用
    // 统计单词的数量，将结果保存到MySQL中
    /**
     * MySQL的建表语句
     * CREATE TABLE `word_cnt` (
     * `word` varchar(255) NOT NULL,
     * `cnt` int(255) DEFAULT NULL,
     * PRIMARY KEY (`word`)
     * ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
     */
    sc
      .textFile("Spark/data/words.txt")
      .flatMap(_.split(","))
      .groupBy(word => word)
      .map(kv => (kv._1, kv._2.size))
      // 每一条数据会执行一次foreach 如果需要同外部系统建立连接
      // 相当于每条数据都要创建一次连接，效率很低
      .foreach(kv => {
        // 连接只能在算子内部创建，因为连接不能被序列化
        val conn: Connection = DriverManager.getConnection("jdbc:mysql://master:3306/student", "root", "123456")
        println("创建了1次连接")
        conn.createStatement().execute("truncate table word_cnt")
        // 创建一个statement
        val pSt: PreparedStatement = conn.prepareStatement("insert into word_cnt(word,cnt) values(?,?)")
        // 设置参数
        pSt.setString(1, kv._1)
        pSt.setInt(2, kv._2)
        // 执行插入语句
        pSt.execute()

        // 关闭连接
        pSt.close()
        conn.close()


      })

    /**
     * mapPartitions vs foreachPartition 如何选择？
     * 需要跟外部系统建立连接:
     * 1、如果需要获取数据 --> mapPartitions
     * 2、如果需要将数据保存至外部系统 --> foreachPartition
     */

    sc
      .textFile("Spark/data/words.txt")
      .flatMap(_.split(","))
      .groupBy(word => word)
      .map(kv => (kv._1, kv._2.size))
      // 对每个分区进行依次处理 将每个分区的数据转换成迭代器
      .foreachPartition(iter => {
        val conn: Connection = DriverManager.getConnection("jdbc:mysql://master:3306/student", "root", "123456")
        println("创建了一次连接")

        conn.createStatement().execute("truncate table word_cnt")

        // 创建一个statement
        val pSt: PreparedStatement = conn.prepareStatement("insert into word_cnt(word,cnt) values(?,?)")
        // 处理每个分区的数据
        iter.foreach(kv => {
          // 设置参数
          pSt.setString(1, kv._1)
          pSt.setInt(2, kv._2)
          // 将一个分区作为一个批次 使用批量插入
          pSt.addBatch()
        })
        // 执行批量插入
        pSt.executeBatch()

        // 关闭连接
        pSt.close()
        conn.close()
      })
  }

}
